import json

from pyflink.common import SimpleStringSchema, WatermarkStrategy
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer


# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

source = KafkaSource.builder() \
    .set_bootstrap_servers("master:9092") \
    .set_topics("students_partition_age") \
    .set_group_id("my-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

student_ds = env.from_source(source, WatermarkStrategy.no_watermarks(), 'Kafka Source')


def map_func(line):
    clazz = json.loads(line)['clazz']
    return clazz,1


clazz_ds = student_ds.map(map_func)

clazz_ds.key_by(lambda x:x[0]).sum(1).print()

env.execute()